package com.sanq.product.mcgs.kafka.producer;

import com.sanq.product.mcgs.kafka.producer.listener.CommonListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.listener.MessageListener;

import javax.annotation.Resource;

public class KafkaMessageListenerImpl implements MessageListener<String, String> {

    @Value("${kafka.common.topic}")
    public String commonTopic;

    @Resource
    private CommonListener commonListener;

    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord) {
        String topic = consumerRecord.topic();
        String data = consumerRecord.value();

        if (commonTopic.equals(topic)) {
            commonListener.run(data);
        }
    }
}
